package org.leiyang.mq;

import com.google.common.collect.Lists;
import org.leiyang.common.dtos.ConnectedMsgServerInfo;
import org.leiyang.common.dtos.MsgDTO;
import org.leiyang.common.dtos.MsgServerInfo;
import org.leiyang.common.entities.MsgOneToOne;
import org.leiyang.common.entities.User;
import org.leiyang.common.holder.MsgServerAndClientHolder;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.leiyang.service.feign.RemoteService;
import org.leiyang.service.impls.RemoteCallImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.Collections;
import java.util.Map;
import java.util.Objects;

import static org.leiyang.common.util.Constants.MSG_SER_RSOCKET_INFO;
import static org.leiyang.common.util.Constants.OFFLINE_MSGS_KEY;

/**
 * 如果客户端在线，根据消息目标客户找到对应的msgServer，进行转发
 */
@Component("onlineMsgConsumer")
public class OnlineMsgConsumer implements StreamListener<String, MapRecord<String, String, String>> {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    public static final String topicName ="online_msg_topic";
    public static final String groupId = "online_msg_group";
    @Value("${onlineMsg.consumer.id}")
    private String onlineMsgConsumerId;
    private final ReactiveRedisCache redisCache;
    private final MsgServerAndClientHolder msgServerAndClientHolder;
    private final RemoteService remoteService;
    private final RemoteCallImpl remoteCall;

    public OnlineMsgConsumer(ReactiveRedisCache redisCache, MsgServerAndClientHolder msgServerAndClientHolder,
                             RemoteService remoteService, RemoteCallImpl remoteCall) {
        this.redisCache = redisCache;
        this.msgServerAndClientHolder = msgServerAndClientHolder;
        this.remoteService = remoteService;
        this.remoteCall = remoteCall;
    }

    /**
     * 目标用户如果在线，则转发到对应的msgServer；
     * 如果不在线，则将该消息视为离线消息存储到redis里
     * @param message never {@literal null}.
     */
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        try {
            Map<String, String> msgMap = message.getValue();
            logger.info("收到在线消息: {}", msgMap);
            MsgOneToOne msg = MsgOneToOne.convertFromMap(msgMap);
            if (Objects.equals(msg.getFromUserId(), -1L)) {
                logger.error("消费者({})消费到的MsgOneToOne无效: {}", onlineMsgConsumerId, msg);
                return;
            }
            //业务处理
            Mono<MsgDTO> msgDTOMono = remoteService.listUsersByIds(String.valueOf(msg.getToUserId()) + "," + msg.getFromUserId())
                    .collectMap(User::getId)
                    .map(userMap -> MsgDTO.convertFromMsgOneToOne(msg, userMap));
            //发送消息
            msgServerAndClientHolder.confirmUserOnline(msg.getToUserId())
                .zipWith(msgDTOMono)
                .flatMap(tuple2 -> {
                    ConnectedMsgServerInfo connectedMsgServerInfo = tuple2.getT1();
                    MsgServerInfo msgServerInfo = connectedMsgServerInfo.getMsgServerInfo();
                    return redisCache.getCacheObject(MSG_SER_RSOCKET_INFO.concat(msgServerInfo.getServerAddress()).concat(":").concat(msgServerInfo.getServerPort()))
                        .map(addPortObj -> {
                            if(Objects.isNull(addPortObj)) {
                                logger.error("未获取到对应的MsgServer地址信息!");
                            }
                            return String.valueOf(addPortObj);
                        })
                        .flatMap(addPort -> remoteCall.batchSendOffLineMsgs(Lists.newArrayList(tuple2.getT2()),
                                connectedMsgServerInfo.getChannelId(),
                                addPort,
                                true)
                        );
                })
                .flatMap(send -> {
                    if(!send) {
                        return redisCache.setCacheSet(OFFLINE_MSGS_KEY + msg.getToUserId(), Collections.singleton(msg))
                                .map(num -> true);
                    }
                    return Mono.just(true);
                }).subscribe(result -> logger.info("已将在线消息({})转发给MsgServer端.", msg.getId()));
        } catch (Exception e) {
            logger.error("消费到Online MsgOneToOne:{} 进行业务处理时出现异常, ", message.getValue(), e);
        } finally {
            //始终会消息应答
            redisCache.createStreamOperations().acknowledge(topicName, groupId, message.getId());
        }
    }
}
